

#include <iostream>

#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <boost/thread/condition.hpp>

#include <xbase_utility.hpp>
#include <dynamic_buffer.hpp>

#include "../lib/protocol.hpp"

namespace p = xbase::protocol;

using namespace xbase;

typedef p::packet_type<p::ROOT_META_REQ>::type req_type;
typedef p::packet_type<p::ROOT_META_REQ>::ptr_type req_ptr;

typedef p::packet_type<p::ROOT_META_RESP>::type resp_type;
typedef p::packet_type<p::ROOT_META_RESP>::ptr_type resp_ptr;

std::size_t count = 0;
std::size_t seq = 0;

typedef boost::asio::ip::tcp::socket s_t;
boost::asio::io_service ios;
boost::asio::ip::tcp::acceptor acceptor(ios);
boost::asio::ip::tcp::endpoint ep;

boost::mutex m2;
boost::condition cond2;
s_t s(ios);
boost::mutex::scoped_lock lock2(m2);


boost::mutex m1;
boost::condition cond1;
s_t server(ios);
boost::mutex::scoped_lock lock1(m1);

boost::mutex m;
boost::condition cond;
std::vector<std::size_t> seqs;
bool s_stopped = false;

void server_recv_thread()
{
	acceptor.accept(server);
	lock1.unlock();

	const std::size_t size = 17;
	char buf[size];
	req_type req;

	bool ok = true;
	boost::system::error_code ec;
	while(ok &&
	      (17 == boost::asio::read(server,
				       boost::asio::buffer(buf),
				       boost::asio::transfer_at_least(size),
				       ec)))
	{
		p::iarchive in(buf, size);
		ok = req.load(in);
		boost::mutex::scoped_lock lock(m);
		seqs.push_back(req.get_sequence_number());
		cond.notify_all();
	}
	s_stopped = true;
	cond.notify_all();
	std::cout << "server_recv_thread quit: " << ec
		  << " - "
		  << ec.message() << std::endl;
}

void server_send_thread()
{
	boost::mutex::scoped_lock lock(m1);
	
	resp_type resp;
	
	resp.body().m_server_address.m_hostname = "127.0.0.1";
	resp.body().m_server_address.m_port = 20081;
	resp.body().m_server_address.m_startcode = 807060504030201ULL;
	
	simple_buffer buffer;
	boost::system::error_code ec;
	while(! ec)
	{
		boost::mutex::scoped_lock lock(m);
		while(seqs.empty() && (!s_stopped))
			cond.wait(lock);
		if(s_stopped)
			break;
		resp.set_sequence_number(seqs.back());
		seqs.pop_back();
		lock.unlock();
		
		p::oarchive out(buffer.prepare(1024 * 1024));
		bool ok = resp.save(out);
		assert(ok);
		buffer.commit(out.used_buffer_size());
		boost::asio::write(server, buffer.data(), boost::asio::transfer_all(), ec);
		buffer.clear();
	}
	std::cout << "server_send_thread quit: " << ec
		  << " - "
		  << ec.message() << std::endl;
}

void send_thread()
{
	s.connect(ep);
	lock2.unlock();

	simple_buffer buffer;

	for(std::size_t i=0; i < count; ++i)
	{
		req_ptr ptr(new req_type());
		p::oarchive out(buffer.prepare(1024 * 1024));
		ptr->set_sequence_number(seq + i);
		bool ok = ptr->save(out);
		assert(ok);
		buffer.commit(out.used_buffer_size());
		boost::asio::write(s, buffer.data());
		buffer.clear();
	}

	std::cout << "send_thread quit" << std::endl;
}

void send_thread_simple()
{
	s.connect(ep);
	lock2.unlock();

	const std::size_t size = 17;
	char buf[size];
	req_type req;

	for(std::size_t i=0; i < count; ++i)
	{
		p::oarchive out(buf, size);
		req.set_sequence_number(seq + i);
		bool ok = req.save(out);
		assert(ok);
		assert(out.used_buffer_size() == size);
		boost::asio::write(s, boost::asio::buffer(buf));
	}

	std::cout << "send_thread quit" << std::endl;
}


void recv_thread()
{	
	boost::mutex::scoped_lock lock(m2);

	simple_buffer buffer;

	buffer.prepare(1024 * 1024);
	std::size_t i=0;
	for(; i < count;)
	{
		std::size_t bytes = s.read_some(buffer.prepare(buffer.left_space()));
		buffer.commit(bytes);

		do {
			const char *begin = boost::asio::buffer_cast<const char*>(buffer.data());
			const char *end = begin + boost::asio::buffer_size(buffer.data());
			std::size_t id, bytes_to_read = 0;
			boost::system::error_code ec;
			p::validate(begin,
				    &end,
				    &bytes_to_read,
				    &id,
				    ec);
			if(ec)
			{
				if((ec != error::header_incomplete) &&
				   (ec != error::packet_incomplete))
				{
					std::cout << "validate failed: " << ec << " - " << ec.message() << std::endl;
					goto QUIT;
				}
				break; // continue read
			}

			std::size_t size = (end - begin);
			p::iarchive in(begin, end);
			resp_ptr ptr(new resp_type());
			bool ok = ptr->load(in);
			if(! ok)
			{
				std::cout << "decode failed: "
					  << dump_string_hex(begin, boost::asio::buffer_size(buffer.data()))
					  << std::endl;
				goto QUIT;
			}
			buffer.consume(size);

			++i;
			std::cout << "resp: " << ptr->to_string()
				  << std::endl;
		} while(buffer.size() > 0);
	}

QUIT:
	if(i != count)
	{
		std::cout << "left data: \n";
		boost::system::error_code ec;
		boost::asio::mutable_buffer b = buffer.prepare(1024 * 1024);
		while((s.read_some(boost::asio::buffer(b), ec) == boost::asio::buffer_size(b)) &&
		      (! ec))
			std::cout << dump_string_hex(boost::asio::buffer_cast<const char*>(b),
						     boost::asio::buffer_size(b));
		std::cout << "\n";
	}

	std::cout << "recv_thread quit" << std::endl;
}

void recv_thread_simple()
{
	boost::mutex::scoped_lock lock(m2);

	const std::size_t headsize = 13;
	char headbuf[headsize];

	simple_buffer buf;

	p::packet head;
	//resp_type resp;
	for(std::size_t i=0; i < count; ++i)
	{
		const std::size_t hs = boost::asio::read(s, boost::asio::buffer(headbuf));
		assert(hs == headsize);
		p::iarchive headin(headbuf, headsize);
		bool headok = head.load(headin);
		if(! headok)
		{
			std::cout << "decode header failed\n";
			break;
		}
		const std::size_t ds = head.get_data_length();
		const std::size_t rs = boost::asio::read(s,
							 buf.prepare(ds));
		assert(ds == rs);
		buf.commit(rs);
		p::iarchive in(buf.data());
		p::root_meta_response resp;
		bool ok = resp.load(in);
		buf.clear();
		if((! ok) || (in.left_buffer_size() != 0))
		{
			std::cout << "decode " << i << " failed: "
				  << resp.to_string() << ": "
				  << dump_string_hex(boost::asio::buffer_cast<const char*>(buf.data()),
						     ds)
				  << std::endl;
			std::cout << "left data: \n";
			boost::system::error_code ec;
			boost::asio::mutable_buffer b = buf.prepare(1024 * 1024);
			while((s.read_some(boost::asio::buffer(b), ec) == boost::asio::buffer_size(b)) &&
			      (! ec))
				std::cout << dump_string_hex(boost::asio::buffer_cast<const char*>(b),
							     boost::asio::buffer_size(b));
			std::cout << "\n";
			      
			break;
		}
		std::cout << head.to_string()
			  << resp.to_string() << std::endl;
	}

	std::cout << "recv_thread quit" << std::endl;
}

int main(int argc, char **argv)
{
	if(argc != 7)
	{
		std::cout << "test_async_root_meta_query <ip> <port> <times> <seq> <mode> <simple>" << std::endl;
		return 0;
	}

	boost::asio::ip::tcp::endpoint ep1(
		boost::asio::ip::address::from_string(argv[1]),
		::atoi(argv[2])
		);
	ep = ep1;

	count = ::atoi(argv[3]);
	seq = ::atoi(argv[4]);
	std::string mode = argv[5];
	std::string simple = argv[6];

	bool server_needed = ((mode == "s") || (mode == "cs") || (mode == "sc"));
	bool client_needed = ((mode == "c") || (mode == "cs") || (mode == "sc"));

	boost::thread *s_sender;
	boost::thread *s_recver;
	if(server_needed)
	{
		acceptor.open(ep.protocol());
		acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
		acceptor.bind(ep);
		acceptor.listen();

		s_sender = new boost::thread(server_send_thread);
		s_recver = new boost::thread(server_recv_thread);
	}

	if(client_needed)
	{
		boost::thread c_sender( ((simple == "s") ? send_thread_simple : send_thread) );
		boost::thread c_recver( ((simple == "s") ? recv_thread_simple : recv_thread) );

		c_sender.join();
		c_recver.join();
		s.close();
	}

	if(server_needed)
	{
		s_sender->join();
		delete s_sender;
		s_recver->join();
		delete s_recver;
		server.close();
	}

	return 0;
}
